package io.kiki.sba.registry.client.remoting;


import io.kiki.sba.registry.client.api.Subscriber;
import io.kiki.sba.registry.client.model.SegmentData;
import io.kiki.sba.registry.client.provider.RegisterCache;
import io.kiki.sba.registry.client.provider.SubscriberImpl;
import io.kiki.sba.registry.client.task.ObserverHandler;
import io.kiki.sba.registry.core.model.ReceivedData;
import io.kiki.sba.registry.core.model.Result;
import io.kiki.stack.netty.processor.BizContext;
import io.kiki.stack.netty.userProcessor.SyncUserProcessor;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

@Setter
@Getter
public class ReceivedDataProcessor extends SyncUserProcessor<ReceivedData> {
    private static final Logger logger = LoggerFactory.getLogger(ReceivedDataProcessor.class);

    private RegisterCache registerCache;

    private ObserverHandler observerHandler;


    public ReceivedDataProcessor(RegisterCache registerCache, ObserverHandler observerHandler) {
        this.registerCache = registerCache;
        this.observerHandler = observerHandler;
    }


    @Override
    public Object syncHandleRequest(BizContext bizContext, ReceivedData request) {
        Result result = new Result();
        result.setSuccess(true);
        if (null == request || null == request.getData()) {
            return result;
        }
        try {
            SegmentData segmentData = new SegmentData();
            segmentData.setData(request.getData());
            segmentData.setVersion(request.getVersion());
            segmentData.setSegment(request.getSegment());
            List<String> subscriberRegistIds = request.getSubscriberRegistIds();
            for (String registId : subscriberRegistIds) {
                Subscriber subscriber = registerCache.getRegistIdToSubscriberMap().get(registId);
                if (null == subscriber) {
                    continue;
                }

                if (subscriber instanceof SubscriberImpl) {
                    SubscriberImpl subscriberImpl = (SubscriberImpl) subscriber;
                    subscriberImpl.putReceivedData(segmentData, request.getLocalZone());
                    try {
                        observerHandler.notify(subscriber);
                    } catch (Exception e) {
                        logger.error("[received] add notify task error, dataId: {}, registId: {}", subscriber.getDataId(), subscriber.getRegistId(), e);
                    }
                } else {
                    logger.warn("[received] ignore unknown subscriber type: {}", subscriber.getClass().getName());
                }
            }

            logger.info("[received] receive subscriber data save success, dataId: {} group: {} version: {} data:{} registIds:{}", request.getDataId(), request.getGroup(), request.getVersion(), request.getData(), subscriberRegistIds);
        } catch (Exception e) {
            result.setSuccess(false);
            result.setMessage("");
            logger.info("[received] receive subscriber data save failed, dataId: {} group: {} version: {} data:{}", request.getDataId(), request.getGroup(), request.getVersion(), request.getData(), e);
        }
        return result;
    }


    @Override
    public String interest() {
        return ReceivedData.class.getName();
    }
}
